package com.tzg157.fitness.service.realtime.dws;

import cn.hutool.json.JSONObject;
import com.tzg157.fitness.service.realtime.HiveOnSparkService;
import com.tzg157.fitness.service.realtime.ads.ADSGenderServiceImpl;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQueryException;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;

@Service("dwsGenderService")
public class DWSGenderServiceImpl implements HiveOnSparkService {

    @Resource
    private ADSGenderServiceImpl adsGenderService;
    @Resource
    private volatile ConcurrentHashMap<String,Object> queueMap;

    @Override
    public void save(JSONObject jsonObject, SparkSession spark) {
        Dataset<Row> dataset = spark.sql("select * from fitness.dim_member_info");
        StringBuilder sql = new StringBuilder();
        sql.append("insert overwrite fitness.dws_gender values");
        List<Row> list = dataset.groupBy("gender").count().collectAsList();
        List<Object> mapList = new ArrayList<>();
        list.forEach(row -> {
            System.out.println("~~~~~~~~~~~~~~~~~~~>>>>>" + row);
            sql.append("('").append(row.getString(0).equals("M") ? "男" : "女").append("',").append(row.getLong(1)).append(")").append(",");
            mapList.add(row.toString());
        });
        queueMap.put("adsGender",mapList);
        System.out.println("~~~~~~~~~~~~~~~~~~~>>>>>queueMap:" + queueMap);
        spark.sql(sql.substring(0,sql.length()-1));
        try {
            adsGenderService.sendToMQ(list);
        } catch (StreamingQueryException e) {
            throw new RuntimeException(e);
        } catch (TimeoutException e) {
            throw new RuntimeException(e);
        }
    }
}
